package com.example;

import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @ClassName: leetcode
 * @Description:
 * @Author: csp
 * @Date:
 */
public class KafkaClient {
    public static Logger log = LoggerFactory.getLogger(KafkaClient.class);
   public KafkaConsumer<String,String> init(){
       Properties props = new Properties();
       props.put("bootstrap.servers", "10.168.4.237:9092");
       props.put("group.id", "article");
       props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
       props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
       KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
       consumer.subscribe(Collections.singleton("articlelog"));
       return consumer;
   }

    public static void main(String[] args) {
        KafkaConsumer<String, String> consumer = new KafkaClient().init();
        Map<String, Object> custCountryMap = new ConcurrentHashMap<>();
        while (true){
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10));
            for (ConsumerRecord<String, String> record : records)  //3)
            {
                log.debug("topic = %s, partition = %s, offset = %d,customer = %s, country = %s\n", record.topic(), record.partition(), record.offset(),
                        record.key(), record.value());
                int updatedCount = 1;
                if (custCountryMap.containsKey(record.value())) {
                    updatedCount = (int) custCountryMap.get(record.value()) + 1;
                }
                custCountryMap.put(record.value(), updatedCount);
                JSONObject json = new JSONObject(custCountryMap);
                System.out.println(json.toString());
            }
        }
    }
}
